package com.jt1078.streamserver.handler;

import com.jt1078.streamserver.flv.FuaAssembler;
import com.jt1078.streamserver.manager.JT1078SessionManager;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;

import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 支持音视频混合国标JT1078推流的统一处理器
 */
public class UnifiedPushHandler extends ChannelInboundHandlerAdapter {

    // 用于视频H264 FU-A拼帧。key: sim_channelNo
    private final Map<String, FuaAssembler> videoAssemblerMap = new ConcurrentHashMap<>();

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf buf = (ByteBuf) msg;
        try {
            if (buf.readableBytes() < 21) return;

            int frameHeader = buf.readInt();
            if (frameHeader != 0x30316364) return;

            // SIM卡长度自动判断（6/10字节）
            int simLen = 6;
            int simMarkIdx = buf.readerIndex();
            byte[] simBytes6 = new byte[6];
            buf.getBytes(simMarkIdx, simBytes6);
            String sim = bcdToString(simBytes6).replaceFirst("^0+", "");
            buf.skipBytes(6);
            int channelNo = buf.readUnsignedByte();

            if (channelNo < 1 || channelNo > 8) {
                // 回退重新读10字节
                buf.readerIndex(simMarkIdx);
                byte[] simBytes10 = new byte[10];
                buf.readBytes(simBytes10);
                sim = bcdToString(simBytes10).replaceFirst("^0+", "");
                simLen = 10;
                channelNo = buf.readUnsignedByte();
            }

            int dataType = buf.readUnsignedByte();     // 0:视频, 4:音频, 其它可扩展

            // 读payload
            byte[] payload = new byte[buf.readableBytes()];
            buf.readBytes(payload);

            JT1078SessionManager.JT1078Session session = JT1078SessionManager.getOrCreateSession(sim);
            session.simLen = simLen;
            session.deviceChannel = ctx.channel(); // 用于对讲指令下发

            String key = sim + "_" + channelNo;

            if (dataType == 0) {
                // 视频H264, FU-A重组
                videoAssemblerMap.putIfAbsent(key, new FuaAssembler());
                FuaAssembler assembler = videoAssemblerMap.get(key);
                List<byte[]> naluList = assembler.feed(payload);
                for (byte[] nalu : naluList) {
                    // 广播到所有订阅队列
                    List<BlockingQueue<byte[]>> queues = session.subscribers.get(channelNo);
                    if (queues != null) {
                        for (BlockingQueue<byte[]> q : queues) {
                            if (!q.offer(nalu)) { q.poll(); q.offer(nalu); }
                        }
                    }
                    // 缓存SPS/PPS
                    if (nalu.length > 4) {
                        int offset = (nalu[2] == 1) ? 3 : 4;
                        int nalType = nalu.length > offset ? (nalu[offset] & 0x1F) : -1;
                        if (nalType == 7) session.spsMap.put(channelNo, nalu);
                        else if (nalType == 8) session.ppsMap.put(channelNo, nalu);
                    }
                }
            } else if (dataType == 4) {
                // G711A音频，直接广播
                List<BlockingQueue<byte[]>> queues = session.audioSubscribers.get(channelNo);
                if (queues != null) {
                    for (BlockingQueue<byte[]> q : queues) {
                        if (!q.offer(payload)) { q.poll(); q.offer(payload); }
                    }
                }
            }
            // 其它类型可扩展

        } finally {
            ReferenceCountUtil.release(buf);
        }
    }

    /**
     * BCD码转字符串，不去前导0
     */
    private static String bcdToString(byte[] bcd) {
        StringBuilder sb = new StringBuilder();
        for (byte b : bcd) sb.append((b >> 4) & 0xF).append(b & 0xF);
        return sb.toString();
    }
}
